#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | date_add(1) | cst_ds }}'
name='72小时断更新汇总表'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
echo -e "统计日期\t扫描站点\t发件下一站\t扫描类型\t数量\t">/tmp/${name}_${dt}.txt;
spark-sql --driver-memory 10G --num-executors 40 --executor-memory 20g --executor-cores 4  --conf spark.sql.shuffle.partitions=500 --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=6g -S -e "
--未操作
with tmp_bill_detail as (
  select
     bill_code
    ,if(latest_scan_type = 3,latest_next_station_name,null) as latest_next_station_name --最新扫描站点下一站
    ,if(latest_scan_type = 3,latest_next_station_code,null) as latest_next_station_code
    ,latest_scan_date         --最新操作扫描时间
    ,case when latest_scan_type = 1  then '快件揽收'
          when latest_scan_type = 2  then '入仓扫描'
          when latest_scan_type = 3  then '发件扫描'
          when latest_scan_type = 4  then '到件扫描'
          when latest_scan_type = 5  then '中心到件'
          when latest_scan_type = 6  then '集货到件'
          when latest_scan_type = 7  then '出仓扫描'
          when latest_scan_type = 8  then '代理点收入扫描'
          when latest_scan_type = 9  then '入库扫描'
          when latest_scan_type = 10 then '快件取出扫描'
          when latest_scan_type = 11 then '问题件扫描'
          when latest_scan_type = 12 then '留仓件扫描'
          else null
      end as latest_scan_type
    ,latest_scan_code         --最新操作扫描网点
    ,unix_timestamp() - unix_timestamp(latest_scan_date) as second_diff  --相差秒数
    ,biz_sign_date
    ,dt
  from jms_dwd.dwd_wide_yl_rt_ssmx_bill_detail_hf
  where dt between date_sub('{{ execution_date | cst_ds }}',21) || '-00' and date_add('{{ execution_date | cst_ds }}',1) || '-07'
  and unix_timestamp(date_add('{{ execution_date | cst_ds }}',1) || ' 07:00:00') - unix_timestamp(latest_scan_date) > 259200   --大于72小时且小于240小时
  and latest_scan_date <> '1999-01-01 00:00:00'   --过滤掉没有操作的数据
  and biz_sign_date = '1999-01-01 00:00:00' --过滤掉已经签收的数据
  and collect_out_site_date = '1999-01-01 00:00:00'
  and is_finish = 0    --未完结
  and latest_scan_type is not null
  and bill_code like 'JT%'
)

--全链路操作
, tmp_all_scan as (
  select
    waybill_no
   ,input_site as scan_site
   ,scan_site_code
   ,scan_time
   ,scan_type
   ,null as next_station
   ,null as next_station_code
  from jms_dwd.dwd_tab_barscan_taking_base_hi
  where substr(dt,1,10) between date_sub('{{ execution_date | cst_ds }}',21) and date_add('{{ execution_date | cst_ds }}',1)
  union all
  select
    waybill_no
   ,input_site as scan_site
   ,scan_site_code
   ,scan_time
   ,scan_type
   ,null as next_station
   ,null as next_station_code
  from jms_dwd.dwd_tab_barscan_warehousing_base_hi
  where substr(dt,1,10) between date_sub('{{ execution_date | cst_ds }}',21) and date_add('{{ execution_date | cst_ds }}',1)
  union all
  select
    waybill_no
   ,input_site as scan_site
   ,scan_site_code
   ,scan_time
   ,scan_type
   ,next_station
   ,next_station_code
  from jms_dwd.dwd_tab_barscan_send_base_hi
  where substr(dt,1,10) between date_sub('{{ execution_date | cst_ds }}',21) and date_add('{{ execution_date | cst_ds }}',1)
  union all
  select
    waybill_no
   ,scan_site
   ,scan_site_code
   ,scan_time
   ,scan_type
   ,null as next_station
   ,null as next_station_code
  from jms_dwd.dwd_tab_barscan_arrival_base_hi
  where substr(dt,1,10) between date_sub('{{ execution_date | cst_ds }}',21) and date_add('{{ execution_date | cst_ds }}',1)
  union all
  select
    waybill_no
   ,scan_site
   ,scan_site_code
   ,scan_time
   ,scan_type
   ,null as next_station
   ,null as next_station_code
  from jms_dwd.dwd_tab_barscan_collect_base_hi
  where substr(dt,1,10) between date_sub('{{ execution_date | cst_ds }}',21) and date_add('{{ execution_date | cst_ds }}',1)
  union all
  select
    waybill_no
   ,scan_site
   ,scan_site_code
   ,scan_time
   ,scan_type
   ,null as next_station
   ,null as next_station_code
  from jms_dwd.dwd_tab_barscan_deliver_base_hi
  where substr(dt,1,10) between date_sub('{{ execution_date | cst_ds }}',21) and date_add('{{ execution_date | cst_ds }}',1)
)

--根据运单去重，取最新
, tmp_no1 as (
  select
    t1.bill_code
   ,t1.latest_next_station_name
   ,t1.latest_next_station_code
   ,t1.latest_scan_date
   ,t1.latest_scan_type
   ,t1.latest_scan_code
   ,t2.scan_time
   ,t2.scan_type
   ,t2.next_station
   ,t2.next_station_code
   ,t2.scan_site
   ,t2.scan_site_code
   ,row_number() over(partition by t1.bill_code order by t2.scan_time desc) rn
  from tmp_bill_detail t1
  left join tmp_all_scan t2
         on t1.bill_code = t2.waybill_no
        and t1.latest_scan_date > t2.scan_time
)

,tmp_result as(
  select
    bill_code

   ,latest_scan_type

   ,case when latest_scan_type = '问题件扫描' then scan_site_code
         else latest_scan_code
    end as latest_scan_code

   ,case when latest_scan_type = '问题件扫描' then scan_time
         else latest_scan_date
    end as latest_scan_date

   ,case when latest_scan_type = '问题件扫描' then scan_type
         else latest_scan_type
    end as scan_type

   ,case when latest_scan_type = '问题件扫描' then scan_site
         else t2.name
    end as latest_scan_name

   ,case when latest_scan_type = '问题件扫描' then next_station_code
         else latest_next_station_code
    end as next_station_code --下一站

   ,case when latest_scan_type = '问题件扫描' then next_station
         else latest_next_station_name
    end as next_station  --下一站

  from tmp_no1 t1
  left join jms_dim.dim_network_whole_massage t2
         on t1.latest_scan_code = t2.code
  where rn = 1
)

select
  date_add('{{ execution_date | cst_ds }}',1) as sum_date
 ,max(latest_scan_name) as latest_scan_name
 ,max(next_station) as next_station
 ,scan_type
 ,count(1) as bill_cnt
from tmp_result
where scan_type is not null
group by latest_scan_code,next_station_code,scan_type
;
   " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt